package cn.coisini.test.kafka.component;

import cn.coisini.test.kafka.config.SpringBootKafkaConfig;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class KafkaDataConsumer {
    private String receivedMessage;

    @KafkaListener(topics = SpringBootKafkaConfig.TOPIC_TEST, groupId = SpringBootKafkaConfig.GROUP_ID)
    public void topic_test(List<String> messages, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        for (String message : messages) {
            final JSONObject entries = JSONUtil.parseObj(message);
            System.out.println(SpringBootKafkaConfig.GROUP_ID + " 消费了： Topic:" + topic + ", Message:" + entries.getStr("data"));
            // 保存接收到的消息
            this.receivedMessage = entries.getStr("data");
        }
    }
    public String getReceivedMessage() {
        return receivedMessage;
    }
}

